package com.mazaiting.rabbitmq.executor;

import com.alibaba.fastjson.JSON;
import com.mazaiting.rabbitmq.callback.IMessageCallback;
import com.mazaiting.rabbitmq.config.MqProperties;
import com.mazaiting.rabbitmq.domain.entity.MqMessage;
import com.mazaiting.rabbitmq.service.IMqMessageService;
import com.mazaiting.redisson.service.IRedissonService;
import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.springframework.amqp.core.Message;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 默认的消息执行器
 * Unacked 的消息, mq 重启后会重新发送
 */
@Slf4j
@Setter
@Component
@RequiredArgsConstructor
public class DefaultMessageExecutor implements IMessageExecutor {

    private final IRedissonService redissonService;

    private final IMqMessageService mqMessageService;

    private final MqProperties mqProperties;

    @Override
    public <T> void execute(Message message, Channel channel, Class<T> clazz, IMessageCallback<T> callback) {
        // 获取消息的 ID
//        String messageId = message.getMessageProperties().getHeader(MqConstant.KEY_RETURNED_MESSAGE_CORRELATION);
        String messageId = message.getMessageProperties().getMessageId();
        log.info("默认消息处理器, 消息ID: " + messageId);
        // 根据 ID 获取消息
        MqMessage mqMessage = mqMessageService.getById(messageId);

        // 判断消息是否存在
        if (Objects.nonNull(mqMessage) && !checkCompleteStatus(mqMessage) && !checkRetryCount(mqMessage)) {
            try {
                // 查数据库判断是否需要持久化
                boolean exclusion = mqMessage.getExclusion();
                // 如果不需要则执行回调
                if (exclusion) {
                    // 拿到一个非公平锁
                    RLock lock = null;
                    // 加锁结果
                    boolean lockResult = false;
                    try {
                        // 获取锁
                        lock = redissonService.getLock(messageId);
                        // 尝试获取锁, 获取到锁之后 10s 后释放
                        lockResult = lock.tryLock(-1, mqProperties.getLeaseTime(), TimeUnit.SECONDS);
                        if (lockResult) {
                            handleMessage(message, clazz, callback);
                        } else {
                            log.error("消息 ID = [{}], 加锁失败.", messageId);
                        }
                    } catch (Exception e) {
                        log.error("消息 ID = [{}], 处理异常: {}", messageId, e.getMessage());
                        throw e;
                    } finally {
                        // 加锁成功并且获取到锁
                        if (Objects.nonNull(lock) && lockResult) {
                            lock.unlock();
                        }
                    }
                } else {
                    // 获取内容
                    handleMessage(message, clazz, callback);
                }
                // 设置接收成功
                mqMessage.setReceiveStatus(true);
                // 设置已完成
                mqMessage.setCompleteStatus(true);

                // 手动确认
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (Exception e) {
                e.printStackTrace();
                log.error("消息 ID [{}], 处理失败-失败原因: {}", messageId, e.getMessage());
                // 设置接收失败
                mqMessage.setReceiveStatus(false);
                mqMessage.setErrorMsg(e.getMessage());
                // 当前重试次数
                Integer retryCount = mqMessage.getRetryCount();
                // 计算重试次数
                mqMessage.setRetryCount(++retryCount);
                try {
                    // 获取最大重试次数
                    Integer maxRetryCount = mqMessage.getMaxRetryCount();
                    // 判断消息是否已经处理 - 不需要重试/当前重试次数大于等于最大重试次数
                    if (retryCount >= maxRetryCount && message.getMessageProperties().getRedelivered()) {
                        log.error("消息 ID [{}], 消息已重复处理失败,拒绝再次接收...", messageId);
                        // 设置已完成
                        mqMessage.setCompleteStatus(true);
                        // 拒绝再次接收
                        channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                    } else {
                        log.error("消息 ID [{}], 消息即将再次返回队列处理...", messageId);
                        // 消息返回队列
                        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                    }
                } catch (Exception ex) {
                    // 发送失败
                    log.error("消息 ID [{}], 处理失败: {}", message, ex.getMessage());
                    mqMessage.setErrorMsg(mqMessage.getErrorMsg() + ex.getMessage());
                }
            } finally {
                // 更新数据
                mqMessageService.updateById(mqMessage);
            }
        } else {
            try {
                log.warn("消息已处理, 手动拒绝. 消息 ID: " + messageId);
                // 手动确认拒绝
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } catch (Exception e) {
                log.error("手动拒绝失败, 消息 ID: {}, 异常原因: {}", messageId, e.getMessage());
            }
        }
    }

    /**
     * 校验重试次数
     *
     * @param mqMessage 消息数据库实体
     * @return true: 校验通过; false: 次数超限
     */
    private boolean checkRetryCount(MqMessage mqMessage) {
        // 获取最大重试次数
        Integer maxRetryCount = mqMessage.getMaxRetryCount();
        // 当前重试次数
        Integer retryCount = mqMessage.getRetryCount();
        try {
            // 当前次数大于等于最大重试次数
            if (retryCount >= maxRetryCount) {
                log.error("当前重试次数超过最大重试次数: 当前重试次数: {}, 最大重试次数: {}", retryCount, maxRetryCount);
                return true;
            }
        } catch (Exception e) {
            log.error("校验消息重试失败, 消息 ID: {}, 异常信息: {}", mqMessage.getId(), e.getMessage());
        }
        return false;
    }

    /**
     * 校验完成状态
     *
     * @param mqMessage 消息数据库实体
     * @return true: 已完成; false: 未完成
     */
    private boolean checkCompleteStatus(MqMessage mqMessage) {
        try {
            // 判断消息是否已经完成
            if (mqMessage.getCompleteStatus()) {
                log.info("消息已经处理完成, ID: {}", mqMessage.getId());
                return true;
            }
        } catch (Exception e) {
            log.error("校验消息完成状态失败, 消息 ID: {}, 异常信息: {}", mqMessage.getId(), e.getMessage());
        }
        return false;
    }

    /**
     * 处理消息
     *
     * @param message  消息
     * @param callback 回调
     */
    private <T> void handleMessage(Message message, Class<T> clazz, IMessageCallback<T> callback) throws Exception {
        // 转换消息
        String msg = new String(message.getBody(), StandardCharsets.UTF_8);
        T t = JSON.parseObject(msg, clazz);
        callback.handleMessage(t);
    }
}
